package com.huiyun.project.pay.biz.mq;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huiyun.project.pay.biz.entity.MchNotify;
import com.huiyun.project.pay.biz.handler.callback.impl.AliPayCallbackHandler;
import com.huiyun.project.pay.biz.handler.callback.impl.WxPayCallbackHandler;
import com.huiyun.project.pay.biz.service.MchNotifyService;
import com.huiyun.project.pay.client.consts.PayConst;
import com.huiyun.project.pay.client.vo.NotifyResp;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

/**
 * 本地事务监听器
 *
 * @author 01
 * @date 2019-08-08
 **/
@Slf4j
@RequiredArgsConstructor
@RocketMQTransactionListener(txProducerGroup = PayConst.TX_PRODUCER_GROUP_REFUND)
public class RefundTransactionListener implements RocketMQLocalTransactionListener {


    private final WxPayCallbackHandler wxPayCallbackHandler;

    private final AliPayCallbackHandler aliPayCallbackHandler;

    private final MchNotifyService mchNotifyService;

    /**
     * 用于执行本地事务的方法
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            log.info("MQ支付消息提交本地事务开始");
            NotifyResp notifyResp = (NotifyResp) arg;
            // 执行带有事务注解的方法,下面几步放到一个事务里面
            if (notifyResp.getChannelId().startsWith("WX_")) {
                //1.修改payOrder结果
                //2.增加mchNotice记录
                wxPayCallbackHandler.process(notifyResp);
            } else if (notifyResp.getChannelId().startsWith("ALIPAY_")) {
                aliPayCallbackHandler.process(notifyResp);
            }
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("MQ支付消息本地事务发生异常，消息将被回滚", ExceptionUtils.getStackTrace(e));
            //发生异常向MQ Server发送rollback消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 用于回查本地事务的执行结果
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        String uniqueKey = Convert.toStr(headers.get("uniqueKey"));
        String[] uniqueKeys = Convert.toStr(headers.get("uniqueKey")).split(":");
        log.warn("回查本地事务状态. 事务id: {}", uniqueKey);
        // 按事务id查询日志数据
        int count = mchNotifyService.count(Wrappers.query(new MchNotify().setMchId(uniqueKey).setMchOrderNo(uniqueKeys[1])));
        // 如果能按事务id查询出来数据表示本地事务执行成功，没有数据则表示本地事务执行失败
        if (count == 0) {
            log.warn("本地事务执行失败，事务日志不存在，消息将被回滚. 事务id: {}", uniqueKey);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        return RocketMQLocalTransactionState.COMMIT;
    }
}